package com.supplychain.scm.core.rocketmq.handler;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.supplychain.scm.core.common.jackson.DbyJavaTimeModule;
import com.supplychain.scm.core.common.util.SpringContextHolder;
import com.supplychain.scm.core.rocketmq.annotation.MessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.core.MethodParameter;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.SmartMessageConverter;
import org.springframework.messaging.support.MessageBuilder;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;

/**
 * @author yanhan
 * @version 1.0.0
 * @ClassName AbstractMessageHandler
 * @Description 消息处理
 * @createTime 2022/08/18 14:27:00
 */
@Slf4j
public abstract class AbstractMessageHandler<T> {

	public String getTag(){
		Class<? extends AbstractMessageHandler> clazz = this.getClass();
		if(!clazz.isAnnotationPresent(MessageHandler.class)) {
			throw new UnsupportedOperationException();
		}
		return clazz.getAnnotation(MessageHandler.class).tag();
	}

	/**
	 * 消息处理器
	 * @param body
	 * @return
	 */
	public abstract boolean messageHandler(T body, MessageExt messageExt);

	public final boolean handler(MessageExt message){
		T t = ((T) this.doConvertMessage(message));
		return messageHandler(t,message);
	}

	private Object doConvertMessage(MessageExt messageExt) {
		RocketMQMessageConverter rocketMqMessageConverter = this.getRocketMqMessageConverter();
		Type messageType = this.getMessageType();
		if (Objects.equals(messageType, MessageExt.class)) {
			return messageExt;
		} else {
			String str = new String(messageExt.getBody(), StandardCharsets.UTF_8);
			if (Objects.equals(messageType, String.class)) {
				return str;
			} else {
				// If msgType not string, use objectMapper change it.
				try {
					if (messageType instanceof Class) {
						//if the messageType has not Generic Parameter
						return rocketMqMessageConverter.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType);
					} else {
						//if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third parameter "conversionHint".
						//we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.
						return ((SmartMessageConverter) rocketMqMessageConverter.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), this.getMethodParameter());
					}
				} catch (Exception e) {
					log.warn("convert failed. str:{}, msgType:{}", str, messageType);
					throw new RuntimeException("cannot convert message to " + messageType, e);
				}
			}
		}
	}

	private Type getMessageType() {
		Class<?> targetClass = AopProxyUtils.ultimateTargetClass(this);
		Type matchedGenericInterface = null;
		while (Objects.nonNull(targetClass)) {
			Type genericSuperclass = targetClass.getGenericSuperclass();
			if (genericSuperclass instanceof ParameterizedType &&
					(Objects.equals(((ParameterizedType) genericSuperclass).getRawType(), AbstractBroadcastMessageHandler.class)
							|| Objects.equals(((ParameterizedType) genericSuperclass).getRawType(), AbstractClusterMessageHandler.class))) {
				matchedGenericInterface = genericSuperclass;
				break;
			}
			targetClass = targetClass.getSuperclass();
		}
		if (Objects.isNull(matchedGenericInterface)) {
			return Object.class;
		}
		Type[] actualTypeArguments = ((ParameterizedType) matchedGenericInterface).getActualTypeArguments();
		if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
			return actualTypeArguments[0];
		}
		return Object.class;
	}

	private MethodParameter getMethodParameter() {
		Class<?> targetClass = AopProxyUtils.ultimateTargetClass(this);
		Type messageType = this.getMessageType();
		Class clazz = null;
		if (messageType instanceof ParameterizedType && this.getRocketMqMessageConverter().getMessageConverter() instanceof SmartMessageConverter) {
			clazz = (Class) ((ParameterizedType) messageType).getRawType();
		} else if (messageType instanceof Class) {
			clazz = (Class) messageType;
		} else {
			throw new RuntimeException("parameterType:" + messageType + " of messageHandler method is not supported");
		}
		try {
			final Method method = targetClass.getDeclaredMethod("messageHandler", clazz,MessageExt.class);
			return new MethodParameter(method, 0);
		} catch (NoSuchMethodException e) {
			log.warn("getMethodParameter error => {}", ExceptionUtils.getStackTrace(e));
			throw new RuntimeException("parameterType:" + messageType + " of messageHandler method is not supported");
		}
	}

	private RocketMQMessageConverter getRocketMqMessageConverter() {
		RocketMQMessageConverter rocketMqMessageConverter = null;
		try {
			rocketMqMessageConverter = SpringContextHolder.getBean(RocketMQMessageConverter.class);
		} catch (Exception e) {
			if(log.isDebugEnabled()) {
				log.debug("rocket mq message converter error => {}",ExceptionUtils.getStackTrace(e));
			}
			rocketMqMessageConverter = new RocketMQMessageConverter();
			CompositeMessageConverter compositeMessageConverter = (CompositeMessageConverter) rocketMqMessageConverter.getMessageConverter();
			List<MessageConverter> messageConverterList = compositeMessageConverter.getConverters();
			for (MessageConverter messageConverter : messageConverterList) {
				if (messageConverter instanceof MappingJackson2MessageConverter) {
					MappingJackson2MessageConverter jackson2MessageConverter = (MappingJackson2MessageConverter) messageConverter;
					ObjectMapper objectMapper = jackson2MessageConverter.getObjectMapper();
					DbyJavaTimeModule module = new DbyJavaTimeModule();
					objectMapper.registerModules(module);
					break;
				}
			}
		}
		return rocketMqMessageConverter;
	}
}
